package net.bwie.realtime.douyin.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/05/26 21:36:51
 * @version: 1.0
 */
public class Dwd_Split_flow {
    public static void main(String[] args) throws Exception {
        //1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.设置并行度
        env.setParallelism(1);
        //3.过滤流
        DataStream<String> Orderstream = KafkaUtil.consumerKafka(env, "sales_topic");

//        displayStream.print("master_flow");
        processOrder(Orderstream);

        env.execute("Dwd_Split_flow");
    }

    private static void processOrder(DataStream<String> orderstream) {
//      1 设置清洗 流
        DataStream<String> CleanStream = logCleaned(orderstream);
//        CleanStream.print();
//      2 根据 订单状态 进行过滤
        DataStream<String> logfilter = Logfilter(CleanStream);

//        logfilter.print();

        KafkaUtil.producerKafka(logfilter, "pay-log");

    }

    private static DataStream<String> Logfilter(DataStream<String> cleanStream) {
        OutputTag<String> exction = new OutputTag<String>("dont pay") {
        };

        SingleOutputStreamOperator<String> pay_stream = cleanStream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, Context context, Collector<String> collector) throws Exception {


                if (JSON.parseObject(s).getString("order_status").equals("1")) {

                    collector.collect(s);
                } else {
                    context.output(exction, s);
                }
            }
        });
//        pay_stream.print();

        return pay_stream;
    }

    private static DataStream<String> logCleaned(DataStream<String> orderstream) {

        OutputTag<String> dirtyTag = new OutputTag<String>("dirty-log") {
        };

        SingleOutputStreamOperator<String> cleanedStream = orderstream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, Context context, Collector<String> collector) throws Exception {
                try {
                    JSON.parse(s);
                    collector.collect(s);
                } catch (Exception e) {
                    context.output(dirtyTag, s);
                }
            }
        });

        DataStream<String> dirtyStream = cleanedStream.getSideOutput(dirtyTag);

        KafkaUtil.producerKafka(dirtyStream, "dwd_traffic-dirty-log");

        return cleanedStream;
    }
}
